package com.atguigu.flink.chapter06.watermark;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.chapter05.function.WaterSensorMapFunction;
import com.atguigu.flink.util.AtguiguUtil;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

public class WaterMarkDemo2 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        //设置水印的生成周期
        env.getConfig().setAutoWatermarkInterval(2000);

        env.socketTextStream("hadoop102",8888)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        new WatermarkStrategy<WaterSensor>() {
                            @Override
                            public WatermarkGenerator<WaterSensor> createWatermarkGenerator(
                                    WatermarkGeneratorSupplier.Context context) {
                                return new MyWmkGenerator();
                            }
                        }.withTimestampAssigner((element,ts) -> element.getTs())   //指定事件时间
                )
                .keyBy(WaterSensor::getId)
                //定义窗口：基于事件时间的滚动窗口，窗口的长度:5s
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String s,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        List<WaterSensor> list = AtguiguUtil.toList(elements);
                        out.collect(ctx.window().getStart() + " " + ctx.window().getEnd() + " " + list);
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    public static class MyWmkGenerator implements WatermarkGenerator<WaterSensor>{

        long maxTs = Long.MIN_VALUE + 3000 + 1;

        @Override
        public void onEvent(WaterSensor event, long et, WatermarkOutput output) {
            maxTs = Math.max(maxTs, et);    //找到最大时间戳
            //流中每来一条数据执行一次
            System.out.println("MyWmkGenerator.onEvent");
            output.emitWatermark(new Watermark(maxTs - 3000 -1));
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            System.out.println("MyWmkGenerator.onPeriodicEmit");
            //周期性的执行：把水印植入到流中
            output.emitWatermark(new Watermark(maxTs - 3000 - 1));
        }
    }
}



//public class WaterMarkDemo2 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        //设置水印的生成周期
//        env.getConfig().setAutoWatermarkInterval(2000);
//
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .assignTimestampsAndWatermarks(new WatermarkStrategy<WaterSensor>() {
//                    @Override
//                    public WatermarkGenerator<WaterSensor> createWatermarkGenerator(
//                            WatermarkGeneratorSupplier.Context context) {
//                        return new MyWmkGenerator();
//                    }
//                })
//                .keyBy(WaterSensor::getId)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
//                    @Override
//                    public void process(String s,
//                                        Context ctx,
//                                        Iterable<WaterSensor> elements,
//                                        Collector<String> out) throws Exception {
//                        List<WaterSensor> list = AtguiguUtil.toList(elements);
//                        out.collect(ctx.window().getStart() + " " + ctx.window().getEnd() + " " + list);
//                    }
//                })
//                .print();
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//
//    public static class MyWmkGenerator implements WatermarkGenerator<WaterSensor>{
//
//        long maxTs = Long.MIN_VALUE + 3000 + 1;
//        @Override
//        public void onEvent(WaterSensor event, long et, WatermarkOutput output) {
//            maxTs = Math.max(maxTs, et);     //找到最大时间戳
//            //流中每来一条数据执行一次
//            System.out.println("MyWmkGenerator.onEvent");
//            output.emitWatermark(new Watermark(maxTs - 3000 - 1));
//        }
//
//        @Override
//        public void onPeriodicEmit(WatermarkOutput output) {
//            System.out.println("MyWmkGenerator.onPeriodicEmit");
//            //周期性的执行:把水印植入到流中
//            output.emitWatermark(new Watermark(maxTs - 3000 - 1));
//        }
//    }
//}


//public class WaterMarkDemo2 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        //设置水印的生成周期
//        env.getConfig().setAutoWatermarkInterval(2000);
//
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .assignTimestampsAndWatermarks(
//                        new WatermarkStrategy<WaterSensor>() {
//                            @Override
//                            public WatermarkGenerator<WaterSensor> createWatermarkGenerator(
//                                    WatermarkGeneratorSupplier.Context context) {
//                                return new MyWmkGenerator();
//                            }
//                        }
//                )
//                .keyBy(WaterSensor::getId)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
//                    @Override
//                    public void process(String s,
//                                        Context ctx,
//                                        Iterable<WaterSensor> elements,
//                                        Collector<String> out) throws Exception {
//                        List<WaterSensor> list = AtguiguUtil.toList(elements);
//                        out.collect(ctx.window().getStart() + " " + ctx.window().getEnd() + " " + list);
//                    }
//                })
//                .print();
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//
//    public static class MyWmkGenerator implements WatermarkGenerator<WaterSensor>{
//
//        long maxTs = Long.MIN_VALUE + 3000 + 1;
//        @Override
//        public void onEvent(WaterSensor event, long et, WatermarkOutput output) {
//            maxTs = Math.max(maxTs,et);   //找到最大时间戳
//            //流中每来一条数据就执行一次
//            System.out.println("MyWmkGenerator.onEvent");
//            output.emitWatermark(new Watermark(maxTs - 3000 - 1));
//        }
//
//        @Override
//        public void onPeriodicEmit(WatermarkOutput output) {
//            System.out.println("MyWmkGenerator.onPeriodicEmit");
//            //周期性执行：把水印植入流中
//            output.emitWatermark(new Watermark(maxTs - 3000 - 1));
//        }
//    }
//}